package org.tao.lightningmq.broker.event.spi.listenr;

import com.alibaba.fastjson2.JSON;
import io.netty.util.AttributeKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tao.lightningmq.broker.cache.CommonCache;
import org.tao.lightningmq.broker.event.model.StartSyncEvent;
import org.tao.lightningmq.common.dto.StartSyncRespDTO;
import org.tao.lightningmq.common.enums.BrokerResponseCode;
import org.tao.lightningmq.common.event.Listener;
import org.tao.lightningmq.common.tcp.TcpMsg;

import java.net.InetSocketAddress;

/**
 * @Author lht
 * @date 2025/1/21 - 13:49
 * @description:
 */
public class StartSyncListener implements Listener<StartSyncEvent> {

    private static final Logger LOGGER = LoggerFactory.getLogger(StartSyncListener.class);

    @Override
    public void onReceive(StartSyncEvent event) throws Exception {
        InetSocketAddress inetSocketAddress = (InetSocketAddress) event.getContext().channel().remoteAddress();
        String reqId = inetSocketAddress.getAddress() + ":" + inetSocketAddress.getPort();
        event.getContext().attr(AttributeKey.valueOf("reqId")).set(reqId);
        LOGGER.info("从节点: {} 接入成功", reqId);
        CommonCache.getSlaveChannelMap().put(reqId, event.getContext());
        StartSyncRespDTO startSyncRespDTO = new StartSyncRespDTO();
        startSyncRespDTO.setMsgId(event.getMsgId());
        startSyncRespDTO.setSuccess(true);
        TcpMsg tcpMsg = new TcpMsg(BrokerResponseCode.START_SYNC_SUCCESS.getCode(), JSON.toJSONBytes(startSyncRespDTO));
        event.getContext().writeAndFlush(tcpMsg);
    }
}
